package cn.sciento.transfer.infra.rocketmq.service.impl;

import com.alibaba.fastjson.JSON;
import java.util.Date;
import java.util.Map;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.tools.command.SubCommandException;
import cn.sciento.core.base.BaseConstants;
import cn.sciento.transfer.domain.entity.ProducerConfig;
import cn.sciento.transfer.domain.repository.ProducerConfigRepository;
import cn.sciento.transfer.infra.rocketmq.RocketmqProducersGroupManager;
import cn.sciento.transfer.infra.rocketmq.config.MqConfigProperties;
import cn.sciento.transfer.infra.rocketmq.service.MqProducerService;
import cn.sciento.transfer.infra.rocketmq.utils.RocketmqClientUtils;
import cn.sciento.transfer.infra.rocketmq.vo.MqMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

@Service
public class MqProducerServiceImpl implements MqProducerService {
  private Logger LOGGER = LoggerFactory.getLogger(MqProducerServiceImpl.class);
  
  @Autowired
  private MqConfigProperties mqConfigProperties;
  
  @Autowired
  private RocketmqProducersGroupManager rocketmqProducersGroupManager;
  
  public void createTopic(String serviceCode, String tableName, String produceGroup) throws SubCommandException {
    try {
      RocketmqClientUtils.createTopic(this.mqConfigProperties.getNameSrv(), 
          MqConsumerServiceImpl.createTopicName(serviceCode, tableName));
    } catch (SubCommandException e) {
      if (this.LOGGER.isDebugEnabled())
        this.LOGGER.debug(">>>>>>>>>>>>>CREATE TOPIC FAILED,TOPIC[{}]<<<<<<<<<<<<<", serviceCode + tableName); 
      throw e;
    } 
  }
  
  public void createTopic(String serviceCode, String tableName) throws SubCommandException {
    createTopic(serviceCode, tableName, "TRANSFER-DEFAULT-PRODUCER");
  }
  
  public void deleteTopic(String serviceCode, String tableName, String produceGroup) {
    try {
      RocketmqClientUtils.deleteTopic(this.mqConfigProperties.getNameSrv(), 
          MqConsumerServiceImpl.createTopicName(serviceCode, tableName));
    } catch (SubCommandException e) {
      if (this.LOGGER.isDebugEnabled())
        this.LOGGER.debug(">>>>>>>>>>>>>>NO TOPIC FOUND,TOPIC[{}]<<<<<<<<<<<<<<<<", serviceCode + tableName); 
    } 
  }
  
  public void deleteTopic(String serviceCoe, String tableName) {
    deleteTopic(serviceCoe, tableName, "TRANSFER-DEFAULT-PRODUCER");
  }
  
  public int sendMsg(MqMessage mqMessage, String produceGroup) {
    DefaultMQProducer defaultMQProducer = createProducer(produceGroup);
    this.rocketmqProducersGroupManager.startProducer(produceGroup);
    try {
      String tag = BaseConstants.Flag.YES.equals(mqMessage.getTenantFlag()) ? mqMessage.getData().get(0).get("tenantId").toString() : "null";
      Message message = new Message(MqConsumerServiceImpl.createTopicName(mqMessage.getSourceService(), mqMessage.getTable()), tag, mqMessage.getEventId().toString(), JSON.toJSONBytes(mqMessage, new com.alibaba.fastjson.serializer.SerializerFeature[0]));
      defaultMQProducer.send(message);
      return 1;
    } catch (Exception e) {
      this.LOGGER.error(e.getMessage(), e);
      return 0;
    } 
  }
  
  public int sendMsg(String mqMsg, String produceGroup) {
    MqMessage mqMessage;
    try {
      mqMessage = JSON.parseObject(mqMsg, MqMessage.class);
    } catch (Exception e) {
      this.LOGGER.error(e.getMessage(), e);
      return 0;
    } 
    return sendMsg(mqMessage, produceGroup);
  }
  
  public int sendMsg(String mqMsg) {
    return sendMsg(mqMsg, "TRANSFER-DEFAULT-PRODUCER");
  }
  
  public DefaultMQProducer createProducer(String produceGroup) {
    return this.rocketmqProducersGroupManager.getOrCreateProducer(produceGroup);
  }
  
  public DefaultMQProducer createDefaultProducer() {
    return createProducer("TRANSFER-DEFAULT-PRODUCER");
  }
  
  @Async("stong-transfer-topic-create")
  public void createTopic(ProducerConfig producerConfig, ProducerConfigRepository producerConfigRepository) {
    try {
      createTopic(producerConfig.getServiceCode(), producerConfig.getTableName());
      producerConfig.setTopicGeneratedStatus("S");
    } catch (SubCommandException e) {
      producerConfig.setTopicGeneratedStatus("E");
      producerConfig.setTopicGeneratedMsg(e.getMessage());
    } 
    producerConfig.setTopicGeneratedTime(new Date());
    producerConfigRepository.updateOptional(producerConfig, new String[] { "topicGeneratedMsg", "topicGeneratedStatus", "topicGeneratedTime" });
  }
}
